package net.bwie.realtime.jtp.dws.douyin.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.dws.douyin.log.bean.DealUserRegionMetric4;
import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.functions.DouYinUserRegionFunction4;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class DouYinDealUserRegion4 {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境（与原类保持一致）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);          // 测试阶段并行度1，生产按需调整（建议与Kafka分区数匹配）
        env.enableCheckpointing(3000L); // 3秒一次Checkpoint，保证容错（生产可调整为10-30秒）

        // 2. 读取Kafka数据源（ods_live_event_log，成交事件在该表中）
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "ods_live_event_log");

        // 3. 数据处理：解析→过滤成交事件→窗口聚合→JSON格式化
        DataStream<String> resultStream = handle(kafkaDataStream);

        // 4. 打印结果（测试用，观察指标格式）
        resultStream.print("成交用户地区画像指标→");

        // 5. 数据输出到Doris（与原类一致，需提前创建Doris表）
        DorisUtil.saveToDoris(
                resultStream,
                "douyin_realtime_report",  // 数据库名
                "dws_deal_user_region"     // 表名（存储地区画像指标）
        );

        // 6. 触发作业执行（必须调用，否则无算子执行）
        env.execute("DouYinDealUserRegion3");
    }

    /**
     * 数据处理核心方法（参考原类handle结构，步骤化处理）
     */
    private static DataStream<String> handle(DataStream<String> stream) {
        // 步骤1：解析Kafka JSON数据为EventLog6实体（适配ods_live_event_log表）
        SingleOutputStreamOperator<EventLog6> parsedStream = stream
                .map(jsonStr -> JSON.parseObject(jsonStr, EventLog6.class))  // JSON字符串→实体类
                .filter(event -> event != null);  // 过滤空数据，避免NullPointerException

        // 步骤2：过滤“成交事件”（下单/支付成功，根据业务定义：此处以“支付”为准，更严谨）
        SingleOutputStreamOperator<EventLog6> dealEventStream = parsedStream
                .filter(event -> {
                    // 成交事件：event_type为“支付”（或“下单”，根据业务选择，此处选支付更准确）
                    return "支付".equals(event.getEventType())
                            && event.getRegion() != null; // 过滤无地区信息的无效数据
                });

        // 步骤3：按主播ID分组 + 滑动窗口（与原类策略一致，测试用5秒窗口，生产改为5分钟）
        WindowedStream<EventLog6, Long, TimeWindow> windowStream = dealEventStream
                .keyBy(EventLog6::getAnchorId)  // 分组键：主播ID（按主播维度统计地区分布）
                .window(SlidingProcessingTimeWindows.of(
                        Time.seconds(5),  // 窗口长度：5秒（测试用）→ 生产改为Time.minutes(5)
                        Time.seconds(1)   // 滑动步长：1秒（测试用）→ 生产改为Time.minutes(1)
                ));

        // 步骤4：应用窗口函数，计算地区画像指标（TOP5占比、订单数、UV）
        SingleOutputStreamOperator<DealUserRegionMetric4> metricStream = windowStream
                .apply(new DouYinUserRegionFunction4());  // 自定义窗口函数

        // 步骤5：格式化指标为JSON（适配Doris写入格式，空值处理避免异常）
        // 先过滤出排名前五的数据，再进行JSON格式化
        SingleOutputStreamOperator<String> resultStream = metricStream
                // 对过滤后的TOP5数据进行JSON格式化
                .map(metric -> String.format(
                        "{\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"cur_date\":\"%s\",\"anchor_id\":\"%d\",\"live_room_id\":\"%d\",\"region\":\"%s\",\"deal_user_uv\":\"%d\",\"total_deal_user_uv\":\"%d\",\"region_ratio\":\"%.2f\",\"region_rank\":\"%d\"}",
                        // 空值处理：字符串用空串，数值用0，避免JSON格式错误
                        metric.getWindowStartTime() != null ? metric.getWindowStartTime() : "",
                        metric.getWindowEndTime() != null ? metric.getWindowEndTime() : "",
                        metric.getCurDate() != null ? metric.getCurDate() : "",
                        metric.getAnchorId() != null ? metric.getAnchorId() : 0,
                        metric.getLiveRoomId() != null ? metric.getLiveRoomId() : 0,
                        metric.getRegion() != null ? metric.getRegion() : "",
                        metric.getDealUserUv() != null ? metric.getDealUserUv() : 0,
                        metric.getTotalDealUserUv() != null ? metric.getTotalDealUserUv() : 0,
                        metric.getRegionRatio() != null ? metric.getRegionRatio() : 0.00,
                        metric.getRegionRank() != null ? metric.getRegionRank() : 0
                ));



        return resultStream;
    }
}
